package com.gjy.rabbitmq.resolve;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.MessageProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 可靠消息生产者
 * 提供消息发布确认、重试机制和消息持久化功能
 * 确保消息不丢失
 *
 * @author gjy
 * @version 1.0
 * @since 2025-10-13 14:36:02
 */
public class ReliableMessageProducer {

    private static final Logger logger = LoggerFactory.getLogger(ReliableMessageProducer.class);

    /**
     * RabbitMQ通道
     */
    private final Channel channel;

    /**
     * 未确认消息序列号集合
     */
    private final SortedSet<Long> unconfirmedSet = new TreeSet<>();

    /**
     * 待确认消息缓存
     */
    private final ConcurrentHashMap<Long, MessageInfo> pendingMessages = new ConcurrentHashMap<>();

    /**
     * 消息ID计数器
     */
    private final AtomicLong messageIdCounter = new AtomicLong(0);

    /**
     * 构造可靠消息生产者
     *
     * @param channel RabbitMQ通道
     * @throws IOException 通道操作异常
     */
    public ReliableMessageProducer(Channel channel) throws IOException {
        this.channel = channel;
        // 开启发布确认模式
        channel.confirmSelect();
        setupConfirmListener();
        logger.info("可靠消息生产者初始化完成");
    }

    /**
     * 发送可靠消息
     *
     * @param exchange   交换机名称
     * @param routingKey 路由键
     * @param message    消息内容
     * @throws IOException 消息发送异常
     */
    public void sendReliableMessage(String exchange, String routingKey, String message)
            throws IOException {
        // 生成全局唯一消息ID
        String messageId = "msg_" + messageIdCounter.incrementAndGet() + "_" + System.currentTimeMillis();

        // 构建消息属性，设置持久化和消息ID
        AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                .messageId(messageId)
                .timestamp(new java.util.Date())
                .deliveryMode(2) // 持久化消息
                .build();

        // 获取下一个发布序列号
        long nextPublishSeqNo = channel.getNextPublishSeqNo();

        // 缓存待确认消息信息
        pendingMessages.put(nextPublishSeqNo, new MessageInfo(messageId, exchange, routingKey, message));

        // 添加到未确认集合
        synchronized (unconfirmedSet) {
            unconfirmedSet.add(nextPublishSeqNo);
        }

        // 发布消息
        channel.basicPublish(exchange, routingKey, properties, message.getBytes(StandardCharsets.UTF_8));
        logger.debug("发送消息: messageId={}, exchange={}, routingKey={}", messageId, exchange, routingKey);
    }

    /**
     * 设置确认监听器
     */
    private void setupConfirmListener() {
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) {
                logger.debug("收到确认消息: deliveryTag={}, multiple={}", deliveryTag, multiple);
                handleConfirm(deliveryTag, multiple, true);
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) {
                logger.warn("收到否定确认消息: deliveryTag={}, multiple={}", deliveryTag, multiple);
                handleConfirm(deliveryTag, multiple, false);
            }
        });
    }

    /**
     * 处理消息确认结果
     *
     * @param deliveryTag 交付标签
     * @param multiple    是否批量确认
     * @param isAck       是否为确认
     */
    private void handleConfirm(long deliveryTag, boolean multiple, boolean isAck) {
        synchronized (unconfirmedSet) {
            if (multiple) {
                // 批量确认处理
                unconfirmedSet.headSet(deliveryTag + 1).forEach(tag -> {
                    MessageInfo messageInfo = pendingMessages.remove(tag);
                    if (messageInfo != null) {
                        if (isAck) {
                            onMessageConfirmed(messageInfo);
                        } else {
                            onMessageFailed(messageInfo);
                        }
                    }
                });
                unconfirmedSet.headSet(deliveryTag + 1).clear();
            } else {
                // 单条确认处理
                unconfirmedSet.remove(deliveryTag);
                MessageInfo messageInfo = pendingMessages.remove(deliveryTag);
                if (messageInfo != null) {
                    if (isAck) {
                        onMessageConfirmed(messageInfo);
                    } else {
                        onMessageFailed(messageInfo);
                    }
                }
            }
        }
    }

    /**
     * 消息确认成功回调
     *
     * @param messageInfo 消息信息
     */
    private void onMessageConfirmed(MessageInfo messageInfo) {
        logger.info("消息确认成功: messageId={}", messageInfo.getMessageId());
        // 可以在这里更新数据库状态等
    }

    /**
     * 消息确认失败回调
     *
     * @param messageInfo 消息信息
     */
    private void onMessageFailed(MessageInfo messageInfo) {
        logger.warn("消息确认失败，需要重发: messageId={}", messageInfo.getMessageId());
        retrySendMessage(messageInfo);
    }

    /**
     * 重发消息
     *
     * @param messageInfo 消息信息
     */
    private void retrySendMessage(MessageInfo messageInfo) {
        try {
            // 限制重试次数为3次
            if (messageInfo.getRetryCount() < 3) {
                messageInfo.incrementRetryCount();
                pendingMessages.put(channel.getNextPublishSeqNo(), messageInfo);
                channel.basicPublish(messageInfo.getExchange(), messageInfo.getRoutingKey(),
                        MessageProperties.PERSISTENT_TEXT_PLAIN,
                        messageInfo.getMessage().getBytes(StandardCharsets.UTF_8));
                logger.info("消息重发成功: messageId={}", messageInfo.getMessageId());
            } else {
                logger.error("消息重试次数已达上限，丢弃消息: messageId={}", messageInfo.getMessageId());
                // 记录到死信队列或告警系统
            }
        } catch (IOException e) {
            logger.error("重发消息失败: messageId={}", messageInfo.getMessageId(), e);
        }
    }

}
